/**
 * Copyright 2020 Inc.
 **/
package com.myz.distributed.transaction.rocketmq.producer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.myz.distributed.transaction.rocketmq.common.TransactionService;
import com.myz.distributed.transaction.rocketmq.common.TransferRecordEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;

import javax.annotation.Resource;

/**
 * @author maoyz0621 on 2020/8/12
 * @version v1.0
 */
@RocketMQTransactionListener(txProducerGroup = "${rocketmq.transaction.group.mysql}")
@Slf4j
public class MySqlonsumerTransaction implements RocketMQLocalTransactionListener {

    @Resource
    TransactionService transactionService;
    @Resource
    ObjectMapper objectMapper;

    /**
     * 事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        log.info("=========================== executeLocalTransaction ============================");
        RocketMQLocalTransactionState state = RocketMQLocalTransactionState.UNKNOWN;
        String id = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        try {
            TransferRecordEntity transferRecord = objectMapper.readValue((byte[])msg.getPayload(), TransferRecordEntity.class);
            boolean isCommit = transactionService.doTransfer(transferRecord.getFromUserId(), transferRecord.getToUserId()
                    , transferRecord.getChangeMoney(), transferRecord.getRecordNo(), id);
            if (isCommit) {
                // 提交事务，它允许消费者消费此消息
                state = RocketMQLocalTransactionState.COMMIT;
            } else {
                // 回滚事务，它代表该消息将被删除，不允许被消费
                state = RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return state;
    }

    /**
     * 事务状态回查
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        log.info("=========================== checkLocalTransaction ============================");
        String id = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        RocketMQLocalTransactionState state = RocketMQLocalTransactionState.UNKNOWN;
        try {
            boolean isCommit = transactionService.checkTransferStatus(id);
            if (isCommit) {
                state = RocketMQLocalTransactionState.COMMIT;
            } else {
                state = RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return state;
    }
}